package data

import (
	"context"
	"encoding/json"
	"log"
	"time"

	v1 "user/api/ws/v1"

	"github.com/rabbitmq/amqp091-go"
)

type RabbitMQ struct {
	conn  *amqp091.Connection
	ch    *amqp091.Channel
	queue string
}

// NewRabbitMQ 创建 RabbitMQ 客户端
func NewRabbitMQ(addr, queue string) (*RabbitMQ, error) {
	conn, err := amqp091.Dial(addr)
	if err != nil {
		return nil, err
	}
	ch, err := conn.Channel()
	if err != nil {
		return nil, err
	}
	_, err = ch.QueueDeclare(queue, true, false, false, false, nil)
	if err != nil {
		return nil, err
	}
	return &RabbitMQ{conn: conn, ch: ch, queue: queue}, nil
}

// PublishMessage 异步发布消息到队列
func (r *RabbitMQ) PublishMessage(ctx context.Context, msg *v1.MessageItem) error {
	data, err := json.Marshal(msg)
	if err != nil {
		return err
	}
	go func() {
		ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
		defer cancel()
		_ = r.ch.PublishWithContext(ctx, "", r.queue, false, false, amqp091.Publishing{
			ContentType: "application/json",
			Body:        data,
		})
	}()
	return nil
}

// ConsumeMessages 消费消息，回调处理
func (r *RabbitMQ) ConsumeMessages(handler func(*v1.MessageItem)) error {
	msgs, err := r.ch.Consume(r.queue, "", true, false, false, false, nil)
	if err != nil {
		return err
	}
	go func() {
		for d := range msgs {
			var msg v1.MessageItem
			if err := json.Unmarshal(d.Body, &msg); err == nil {
				handler(&msg)
			} else {
				log.Printf("RabbitMQ message unmarshal error: %v", err)
			}
		}
	}()
	return nil
}

// Close 关闭连接
func (r *RabbitMQ) Close() {
	_ = r.ch.Close()
	_ = r.conn.Close()
}
